package cn.com.dhcc.ddi.task.service.load;

import cn.com.dhcc.app.core.CoreEnv.ParaGetTool;
import cn.com.dhcc.app.core.dao.SimpleDaoForOracle;
import cn.com.dhcc.app.core.exception.DaoException;
import cn.com.dhcc.app.core.exception.NiceException;
import cn.com.dhcc.app.core.exception.ServiceException;
import cn.com.dhcc.app.core.handler.dbhandler.SingleTransationCircleWithOutResult;
import cn.com.dhcc.app.core.handler.threadpoolhandler.ThreadPool;
import cn.com.dhcc.app.core.service.Log;
import cn.com.dhcc.app.core.tag.CommonStaticCode;
import cn.com.dhcc.app.core.util.StrUtil;
import cn.com.dhcc.app.pub.core.consts.DdiProperties;
import cn.com.dhcc.ddi.datasource.vo.DataSourceVo;
import cn.com.dhcc.ddi.task.dao.TaskExecuteLogDao;
import cn.com.dhcc.ddi.task.service.ChangeVo;
import cn.com.dhcc.ddi.task.service.ChangeVo.ConvertInfo;
import cn.com.dhcc.ddi.task.vo.TaskExecuteLog;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * 接收sql语句
 * @Date：2016/11/18 14:12
 * @author：张洪涛
 * @版权所有：东华软件
 * @版本：1.0
 */
@Service
public class ExecuteChangeDbDataImpl {
	private static Logger logger = Log.getLogger("LOAD_LOG");
	private boolean allowPartError = ParaGetTool.getPara(Boolean.class,"recv.sql.allow_part_error", true);//默认允许部分失败
	@Autowired
	private TaskExecuteLogDao taskExecuteLogDao;
	
	static{
		//创建一个错误日志 打印线程
		ThreadPool.createPool("ERROR_SQL", 1, 1, Integer.MAX_VALUE);
	}
	
	/**
	 * 日志写入线程
	 * @author Administrator
	 *
	 */
	private static class ErrorSqlThread extends Thread{
		private File file;
		private String msg;
		private Exception e;
		ErrorSqlThread(File file, String msg, Exception e) {
			super();
			this.file=file;
			this.msg = msg;
			this.e = e;
		}
		public void run() {
			if (!file.getParentFile().exists()) {
				file.getParentFile().mkdirs();
			}
			BufferedWriter bw=null;
			try {
				FileWriter writer = new FileWriter(file,true);
				bw=new BufferedWriter(writer); 
				bw.write(msg); 
				bw.newLine();
				bw.write("出错原因:"); 
				bw.newLine();
				int count=3;
				Throwable t=e.fillInStackTrace();
				while (count>0) {
					if (t!=null) {
						bw.write("      "+t.toString());
						bw.newLine();
						if (t.getCause()!=null) {
							t=t.getCause().fillInStackTrace();
						}else{
							break;
						}
						count--;
					}else{
						break;
					}
				}
				bw.write("**********************************"); 
				bw.newLine();
				bw.flush();
			} catch (IOException e) {
				//
			} finally{
				if (bw!=null) {
					try {
						bw.close();
					} catch (IOException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
				}
			}
			
		}
		
	}

	public Boolean executeChangeSql(DataSourceVo dataSource,
			final List<String> changeSqls) throws NiceException {
		final SimpleDaoForOracle daoForOracle = new SimpleDaoForOracle();
		daoForOracle.initDataSource(dataSource.getDataSourceConfVo().getDriverName(), dataSource.getDataSourceConfVo().getUrl(), dataSource.getDataSourceConfVo().getUserName(), dataSource.getDataSourceConfVo().getPassword());
		String desc = "执行增量sql";
		boolean executeOk = false;
		if(allowPartError){
			return doWithOutTransationCircle( daoForOracle, changeSqls);
		}else{
			 executeOk = doInSingleTransationCircle(daoForOracle, changeSqls);
		}
		return executeOk;
	}
	private Boolean doWithOutTransationCircle(final SimpleDaoForOracle daoForOracle, final List<String> changeSqls){
		for (String sql : changeSqls) {
			try {
				daoForOracle.dmlSql("执行增量sql", sql, new Object[]{});
			} catch (Exception e) {
				if (!passThisException(e)) {
					return false;
				}
			}
		}
		return true;
	}

	/**
	 * 数据入库
	 * @param desc
	 * @param dataSource
	 * @param changeSqls
	 * @param convertMap
     * @return
     */
	public Boolean doInSingleTransationCircleByParam(String desc,final DataSourceVo dataSource ,final List<Map<String,List<Object[]>>> changeSqls,final Map<String,Object> convertMap,String loadLogid){
		final SimpleDaoForOracle daoForOracle = new SimpleDaoForOracle();
		daoForOracle.initDataSource(dataSource.getDataSourceConfVo().getDriverName(), dataSource.getDataSourceConfVo().getUrl(), dataSource.getDataSourceConfVo().getUserName(), dataSource.getDataSourceConfVo().getPassword());
		return this.doInSingleTransationCircleByParam(desc, daoForOracle, changeSqls,convertMap,loadLogid);
	}
	
	public Boolean doInSingleTransationCircle(final DataSourceVo dataSource ,final List<String> changeSqls){
		
		final SimpleDaoForOracle daoForOracle = new SimpleDaoForOracle();
		daoForOracle.initDataSource(dataSource.getDataSourceConfVo().getDriverName(), dataSource.getDataSourceConfVo().getUrl(), dataSource.getDataSourceConfVo().getUserName(), dataSource.getDataSourceConfVo().getPassword());
		return this.doInSingleTransationCircle(daoForOracle, changeSqls);
    }
	
	private Object[] addList(Object[] oldArr,List<Integer> list){
		Object[] newArr=null;
		if (list!=null && !list.isEmpty()) {
			Object[] temp=new Object[list.size()];
			
			for (int i=0;i<list.size();i++) {
				temp[i]=oldArr[list.get(i)];
			}
			newArr=ArrayUtils.addAll(oldArr, temp);
		}
		return newArr;
	}
	
	

	private String getStr(Object[] args){
		if (args==null || args.length<=0) {
			return "";
		}
		StringBuilder sb = new StringBuilder();
		for (int i = 0; i < args.length; i++) {
			Object object = args[i];
			if (i!=args.length-1) {
				if (object!=null && object instanceof Date) {
					sb.append(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format((Date) object)).append(",");
				}else{
					sb.append(object).append(",");
				}
			}else{
				sb.append(object);
			}
		}
		
		return sb.toString();
		
	}
	
	/**
	 * 超过多长时间打印日志
	 * @param end
	 * @param start
	 * @param interval
	 */
	private void logTime(String prestr,Date end,Date start,long interval){
		if (end.getTime()-start.getTime()>=interval) {
			cn.com.dhcc.app.core.handler.loggerhandler.Logger.setToDb(false);
//			cn.com.dhcc.app.core.handler.loggerhandler.Logger.setToLog(true);
			logger.info(prestr+"【"+(end.getTime()-start.getTime())+"】毫秒");
		}
	}
	private Boolean doInSingleTransationCircleByParam(String desc, final SimpleDaoForOracle daoForOracle, final List<Map<String, List<Object[]>>> changeSqls, final Map<String, Object> infoMap , final String loadLogid){
		//记录错误sql的目录
		final File errorFile=new File(DdiProperties.APP_UPLOAD_FILEDATA_PATH,DdiProperties.APP_EXECUTE_ERROR);
		final String taskLogId = (String) infoMap.get("task_log_id");
		final String id=(String) infoMap.get("id");//解析唯一标识
		try {
			final TaskExecuteLog loadLog= taskExecuteLogDao.get(loadLogid);
			if (!errorFile.exists()) {
				errorFile.mkdirs();
			}
			daoForOracle.doInSingleTransationCircle("执行增量sql", new SingleTransationCircleWithOutResult() {
				@Override
				public void actionInCircle() throws RuntimeException {
					
					boolean over=(Boolean) infoMap.get("over");//是否解析完毕

				        //
					int total=0,insertCount=0,updateCount=0,errorCount=0;
					Date start = null;
						for (Iterator<Map<String,List<Object[]>>> iterator = changeSqls.iterator(); iterator
								.hasNext();) {
							Map<String,List<Object[]>> tMap =  iterator.next();
							Set<String> sqls=tMap.keySet();//
							for (String sql : sqls) {
								List<Object[]> argsList=tMap.get(sql);//得到参数
								for (Object[] args : argsList) {
									try {
										total=total+1;//总共执行多少条，计数
										start=new Date();
//										cn.com.dhcc.app.core.handler.loggerhandler.Logger.setToDbAndLog(false);
									    daoForOracle.dmlSql("执行sql", sql, args);
									    if(sql.trim().startsWith("insert into")){//统计insert
									    	logTime("执行第"+total+"条insert耗时",new Date(),start,100);
									    	insertCount=insertCount+1;
									    }
									} catch (Exception e) {
										logTime("执行第"+total+"条insert失败["+sql+"]，失败耗时",new Date(),start,30);
										if(passThisException(e)){
											errorCount=errorCount+1;
//											cn.com.dhcc.app.core.handler.loggerhandler.Logger.setToDbAndLog(true);
											File log=new File(errorFile,taskLogId+".error");
											try {
												ThreadPool.execute("ERROR_SQL", new ErrorSqlThread(log, "第"+total+"条sql出错，sql["+sql+"],args["+getStr(args)+"]", e));
											} catch (ServiceException e1) {
												logger.error("记录执行sql错误日志出错, sql为：" + sql + "args为：" + getStr(args), e1);
											}
											logger.error("执行sql发生异常,"+e.getMessage()+",该异常已忽略",e);
										}else{
											//logger.error("执行sql发生异常,"+e.getMessage()+",尝试更新语句");
											//如果新增失败，异常为主键冲突，则进行更新操作
											start=new Date();
											@SuppressWarnings("unchecked")
											Map<String,ConvertInfo> convertMap=(Map<String, ChangeVo.ConvertInfo>) infoMap.get("convertMap");
											if(convertMap.containsKey(sql)){//如果新增失败，改为update
												ConvertInfo convertInfo=convertMap.get(sql);//update语句
												Object[] newArgs=addList(args, convertInfo.getIndex());//改变参数，新增主键元素
												//logger.error("由于执行失败，尝试更新，更新sql【"+convertInfo.getUpdateSql()+"】,参数值为【"+printArray(newArgs)+"】");
											    //System.err.println("由于现在失败，改为更新，更新sql【"+convertInfo.getUpdateSql()+"】,参数值为【"+args+"】");
												try {
													//logTime("执行第"+total+"条insert失败，转update耗时",new Date(),start,5);
													start=new Date();
													daoForOracle.dmlSql("转换成update语句", convertInfo.getUpdateSql(), newArgs);
													logTime("执行第"+total+"条insert失败，执行转换的update["+convertInfo.getUpdateSql()+"]耗时",new Date(),start,100);
													updateCount=updateCount+1;
												} catch (DaoException e1) {
													errorCount=errorCount+1;
													//logger.error("执行转换成update语句时发生异常,"+e1.getMessage());
//													cn.com.dhcc.app.core.handler.loggerhandler.Logger.setToDbAndLog(true);
													if (!errorFile.exists()) {//如果不存在创建目录
														errorFile.mkdirs();
													}
													File log=new File(errorFile,id+".error");
													
													try {
														ThreadPool.execute("ERROR_SQL", new ErrorSqlThread(log, "第"+total+"条sql出错，sql["+convertInfo.getUpdateSql()+"],args["+getStr(newArgs)+"]", e1));
													} catch (ServiceException e2) {
														logger.error("记录执行sql错误日志出错, sql为：" + convertInfo.getUpdateSql() + "args为：" + getStr(newArgs), e2);
													}
													
													logger.info("执行第"+total+"条insert失败，转update执行失败");
												}
											}
										}
									}
								}
							}
						}
						String printString="";
						if (insertCount>0) {
							printString = printString+"insert执行成功【"+insertCount+"】条,";
						}
						if (updateCount>0) {
							printString = printString+"update执行成功【"+updateCount+"】条,";
						}
						if (errorCount>0) {
							printString = printString+"执行失败【"+errorCount+"】条,";
						}
					loadLog.setTotal(loadLog.getTotal()+total);
					loadLog.setInsertNum(loadLog.getInsertNum() + insertCount);
					loadLog.setUpdateNum(loadLog.getUpdateNum() + updateCount);
					loadLog.setErrorNum(loadLog.getErrorNum() + errorCount);
					try {
						taskExecuteLogDao.update(loadLog);
					} catch (DaoException e) {
						logger.error(e);
					}
//					cn.com.dhcc.app.core.handler.loggerhandler.Logger.setToDbAndLog(true);	
				   logger.info("============当前线程id【"+Thread.currentThread().getId()+"-"+Thread.currentThread().getName()+"】,执行sql结果,共【"+total+"】条，"+printString);
				   //进行一些数据的统计更新
				}
				
			});
			return true;
		} catch (Exception e) {
			if(e instanceof DaoException && e.getMessage().equals("执行增量sql 错误")) {
				//如果因为接收数据库异常中断等原因，导致事务失败，则将次失误所有sql记录错误日志
				logger.error("执行批量sql发生异常,"+e.getMessage()+",该异常已忽略",e);
				//将此次事务所有sql记录到错误日志文件
				File log=new File(errorFile,taskLogId+".error");
				saveBatchErrorSql(log, changeSqls, e);
			}
			return false;
		} 
	}
	
	//批量记录错误sql
	private void saveBatchErrorSql(File log, List<Map<String,List<Object[]>>> changeSqls, Exception e) {
		for (Map<String, List<Object[]>> tMap : changeSqls) {
			Set<String> sqls = tMap.keySet();//
			for (String sql : sqls) {
				List<Object[]> argsList = tMap.get(sql);//得到参数
				for (Object[] args : argsList) {
					try {
						ThreadPool.execute("ERROR_SQL", new ErrorSqlThread(log, "sql出错：sql[" + sql + "],args[" + getStr(args) + "]", e));
					} catch (ServiceException e1) {
						logger.error("记录错误sql出错", e1);
					}
				}
			}
		}
		
		
	}
	
	private Boolean doInSingleTransationCircle( final SimpleDaoForOracle daoForOracle, final List<String> changeSqls){
		try {
			daoForOracle.doInSingleTransationCircle("执行增量sql", new SingleTransationCircleWithOutResult() {
				@Override
				public void actionInCircle() throws RuntimeException {
					for (String sql : changeSqls) {
						try {
							daoForOracle.dmlSql("执行增量sql", sql, new Object[]{});
						} catch (Exception e) {
							//throw new RuntimeException(e);
							logger.error(e);
						}
					}
				}
			});
			return true;
		} catch (Exception e) {
			return false;
		} 
	}
	
	/**
	 * 主键冲突异常要拦截 转为update操作，其他异常pass掉
	 * @param e
	 * @return
	 */
	private boolean passThisException(Exception e){
		String msg = e.getCause().getMessage();
		if(StrUtil.isNotBank(msg)&& (msg.toUpperCase().contains("ORA-00001") || msg.toUpperCase().contains("PRIMARY KEY") || msg.toUpperCase().contains("PRIMARY"))){//唯一性约束
			return false;
		}else{
			logger.error(e);
		}
		return true;
	}
}
